package com.dtkavin.sparkstreaming

import java.net.InetSocketAddress

import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.flume.FlumeUtils
import org.apache.spark.streaming.{Milliseconds, StreamingContext}

/**
  * Created by IntelliJ IDEA.
  * Programmer : John Zn
  * Date : 2016/4/20 0020
  * Time : 00:33
  * Discribtion : 通过streaming这端从flume来pull数据
  * 同时在flume采集机上，需要添加  commons-lang3-3.3.2.jar，scala-library-2.10.5.jar ， spark-streaming-flume-sink_2.10-1.6.1.jar
  * 注意：flume下沉到自己网卡端口，stream从对方网卡端口拉取
  */
class FlumePoll {

}

object FlumePoll {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("FlumePull").setMaster("local[3]")
    val ssc = new StreamingContext(conf, Milliseconds(5000))

    val address = Seq(new InetSocketAddress("192.168.98.51", 8888))//连接不到虚拟机，只能连接到本地虚拟网卡，醉了
    val flumeStream = FlumeUtils.createPollingStream(ssc, address, StorageLevel.MEMORY_AND_DISK)
    val res = flumeStream.flatMap(x => new String(x.event.getBody().array()))
    res.print()

    ssc.start()
    ssc.awaitTermination()
  }
}